sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbWriteSchemaTransformProvider.java (105)

The error output PCollection is being assigned a nested error schema. The variable errorSchema (defined on line 81) already represents the schema for error records (containing the failed row and the error message). Calling ErrorHandling.errorSchema(errorSchema) again wraps this schema unnecessarily, which will lead to a schema mismatch or an unexpectedly nested structure in the error output.

          outputTuple.get(ERROR_TAG).setRowSchema(errorSchema);

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbWriteSchemaTransformProvider.java (127-130)

The conversion from Beam Row to BSON Document is shallow. If the input Row contains nested Row objects or Iterable fields (arrays), row.getValue() will return Beam-specific types that the MongoDB Java driver cannot serialize. This will result in a CodecConfigurationException at runtime. A recursive conversion is required to transform nested Beam Rows into BSON Documents and Iterables into BSON Lists.

sdks/python/apache_beam/yaml/yaml_io.py (744-753)

The row_to_dict function performs a shallow conversion. If the input PCollection contains rows with nested structures (nested rows or lists of rows), _asdict() or dict(row) will not recursively convert those nested Beam Row objects into dictionaries. This will likely cause serialization errors when the data is passed to the underlying mongodbio.WriteToMongoDB transform, as pymongo expects standard Python types. Additionally, avoid using a bare except: block; specify the exception type (e.g., except Exception:) to avoid catching system-exiting exceptions like SystemExit or KeyboardInterrupt.